package extractor;

import data.DataPoint;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.transforms.SimpleFunction;
import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.PCollectionTuple;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Created by emily on 17-12-12.
 */
public class SingleExtractor extends SimpleFunction<KV<String,String>, DataPoint> implements Extractor {

    private static final Logger LOG = LoggerFactory.getLogger(SingleExtractor.class);

    @Override
    public DataPoint apply(KV<String,String> input){

        String[] temp = input.getValue().split(",");
        DataPoint dp = new DataPoint(temp[0], temp[1], temp[2]);
        return dp;
    }



}
